package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo20ForeachPartition {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo20ForeachPartition")

    val sc: SparkContext = new SparkContext(conf)

    // 统计班级人数，将最后的结果保存到MySQL中
    val clazzCntRDD: RDD[(String, Int)] = sc.textFile("spark/data/stu/students.txt")
      .map(line => (line.split(",")(4), 1))
      .reduceByKey(_ + _)

    clazzCntRDD
      // 使用foreach操作同样会出现 连接建立过多的问题 性能比较低
      .foreach(t2 => {
        val clazz: String = t2._1
        val cnt: Int = t2._2

        // 建立连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/stu023", "shujia23", "123456")

        // 创建Statement
        val pSt: PreparedStatement = conn.prepareStatement("insert into clazz_cnt(clazz,cnt) values (?,?)")

        // 设置参数
        pSt.setString(1, clazz)
        pSt.setInt(2, cnt)

        // 如果有大量数据插入时，可以使用批量插入
        pSt.addBatch() // 将每一次插入放入Batch中

        pSt.executeBatch() // 执行批量插入

      })

    // 使用foreachPartition大大减少连接建立的次数
    clazzCntRDD
      .foreachPartition(iter => {
        // 建立连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/stu023", "shujia23", "123456")
        // 创建Statement
        val pSt: PreparedStatement = conn.prepareStatement("insert into clazz_cnt(clazz,cnt) values (?,?)")
        iter.foreach(t2 => {
          val clazz: String = t2._1
          val cnt: Int = t2._2
          // 设置参数
          pSt.setString(1, clazz)
          pSt.setInt(2, cnt)

          // 如果有大量数据插入时，可以使用批量插入
          pSt.addBatch() // 将每一次插入放入Batch中
        })
        pSt.executeBatch() // 执行批量插入
      })
  }

}
